package com.socket.netty;


import com.socket.batch.collector.DataAnalysis;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.util.internal.logging.InternalLoggerFactory;
import io.netty.util.internal.logging.Log4JLoggerFactory;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;

/**
 * Created by kaddddd on 2017/7/4 0004.
 */
public class NettyServer {

    private static final transient Logger log = Logger.getLogger(NettyServer.class);

    private NioEventLoopGroup bossGroup = null;
    private NioEventLoopGroup workerGroup = null;
    private ChannelFuture closeFuture = null;
//    private int acceptThreadNum;
//    private int ioThreadNum;
//    private int port;


    @Value("${nettyServer.host:127.0.0.1}")
    String host;
    @Value("${nettyServer.ioThreadNum:5}")
    private int ioThreadNum;
    @Value("${nettyServer.acceptThreadNum:5}")
    private int acceptThreadNum;
    //内核为此套接口排队的最大连接个数，对于给定的监听套接口，内核要维护两个队列，未链接队列和已连接队列大小总和最大值
    @Value("${nettyServer.backlog:1024}")
    int backlog;
    @Value("${nettyServer.port:8085}")
    int port;

    private NettyServer(int port,int acceptThreadNum,int ioThreadNum) {
        this.port = port;
        this.acceptThreadNum = acceptThreadNum;
        this.ioThreadNum = ioThreadNum;
    }

    public static NettyServer build(int port,int acceptThreadNum,int ioThreadNum) throws Exception {
        NettyServer server = new NettyServer(port,acceptThreadNum,ioThreadNum);
        return server;
    }

    public NettyServer start() throws InterruptedException {

        InternalLoggerFactory.setDefaultFactory(Log4JLoggerFactory.INSTANCE);


        bossGroup = new NioEventLoopGroup(acceptThreadNum, new ThreadFactory() {
            final private AtomicLong THREAD_COUNT = new AtomicLong(0);

            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("rest-boss-thread-" + THREAD_COUNT.getAndIncrement());
                t.setDaemon(false);
                return t;
            }
        });

        workerGroup = new NioEventLoopGroup(ioThreadNum,
                new ThreadFactory() {
                    final private AtomicLong THREAD_COUNT = new AtomicLong(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r);
                        t.setName("csf-rest-io-thread-" + THREAD_COUNT.getAndIncrement());
                        t.setDaemon(false);
                        return t;
                    }
                });

        ServerBootstrap boot = new ServerBootstrap();
        boot.group(bossGroup, workerGroup)
                // nio server socket
                .channel(NioServerSocketChannel.class)
                // server channel参数
                .option(ChannelOption.SO_BACKLOG, 1024)
                //
                .option(ChannelOption.SO_REUSEADDR, true)
                // 关闭keepalive，使用心跳
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                //
                .option(ChannelOption.SO_SNDBUF,1024)
                //
                .option(ChannelOption.SO_RCVBUF,1024)
                .option(ChannelOption.SO_TIMEOUT, 1000*80)
                // child channel参数
                .childOption(ChannelOption.TCP_NODELAY, true)
                // 不优先使用direct memory
                .childOption(ChannelOption.ALLOCATOR, new PooledByteBufAllocator(false))//PooledByteBuf的分配器，通过它来分配一个池化的ByteBuf对象
                // accept端口
                //.localAddress(port)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        byte[] bytes =DataAnalysis.hex2Bytes("5D");
                        ByteBuf byteBuf = Unpooled.copiedBuffer(bytes);
                        //使用分隔符和定长解码器
                        ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, byteBuf));
                        //ch.pipeline().addLast("encoder", new ObjectEncoder());
                        //ch.pipeline().addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                        //具体的业务逻辑的处理
                        ch.pipeline().addLast(
                                // 客户端接收到的是httpResponse响应，所以要使用HttpResponseDecoder进行解码
//                                new HttpResponseDecoder(),
//                                // 客户端发送的是httprequest，所以要使用HttpRequestEncoder进行编码
//                                new HttpRequestEncoder(),
                                //new IdleStateHandler(0,0,RestServerConfigUtils.channelIdleTimeout(),TimeUnit.MILLISECONDS), //
//                                new HttpServerCodec(),
//                                new HttpObjectAggregator(65536),
                                "handler", new DateServerHandler());

                    }
                });

        ChannelFuture f = boot.bind(port).sync();
        f.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    throw new RuntimeException("Rest module start failed bind port:" + port, future.cause());
                }
                log.error("Rest module start successfull!");
            }
        });

        this.closeFuture = f.channel().closeFuture();
        return this;
    }

    public void waitToClose() {
        try {
            this.closeFuture.sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void stop() {
        if (this.bossGroup != null) {
            this.bossGroup.shutdownGracefully();
        }

        if (this.workerGroup != null) {
            this.workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        System.out.println("args = [" + args + "]");
        int port = Integer.parseInt(args[0]);
        int acceptThreadNum = Integer.parseInt(args[1]);
        int ioThreadNum = Integer.parseInt(args[2]);
        System.out.println("--------------------------------------------------------------------------------------");
        System.out.println("                                   启动端口："+args[0]);
        System.out.println("--------------------------------------------------------------------------------------");
        System.out.println("acceptThreadNum = "+acceptThreadNum);
        System.out.println("ioThreadNum = "+ioThreadNum);
        NettyServer.build(port,acceptThreadNum,ioThreadNum).start().waitToClose();
    }
}
